package com.desheng.bigdata.flink.stream.source

import java.io.{BufferedReader, InputStreamReader}
import java.net.Socket

import org.apache.flink.streaming.api.functions.source.SourceFunction
import org.apache.flink.streaming.api.scala._

/**
  * 用户自定义source，来模拟socket
  */
object _02UserDefineSourceFromSocket {
    def main(args: Array[String]): Unit = {
        val env = StreamExecutionEnvironment.getExecutionEnvironment
        val lines = env.addSource(new MySocketSourceFunction("bigdata01", 9999)).setParallelism(2)

        lines.flatMap(_.split("\\s+")).print()

        env.execute(s"${_02UserDefineSourceFromSocket.getClass.getSimpleName}")
    }
}
/*
    通过socket读取数据
    通过扩展SourceFunction而得到的Source，其并行度只能为1
 */
class MySocketSourceFunction(host: String, port: Int) extends SourceFunction[String] {
    /**
      * 资源的构建需要我们在run方法中完成一个分配
      * 当有数据进入之后，才能启动socket连接进行构建，
      * 在run方法外面创建资源，比如socket，会出现NullPointerException
      */
    var socket: Socket = null
    //一旦source发送数据之后，就会调用该run方法
    override def run(ctx: SourceFunction.SourceContext[String]): Unit = {
        socket = new Socket(host, port)
        val br = new BufferedReader(new InputStreamReader(socket.getInputStream))
        var line: String = null
        while((line = br.readLine()) != null) {
            ctx.collect(line)
        }
    }

    //取消读取操作
    override def cancel(): Unit = {
        if(socket != null) {
            socket.close()
        }
    }
}
